package com.atguigu.flink.chapter05.Transform;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/8/9 10:30
 */
public class Flink02_RichMapFunction {
    public static void main(String[] args) throws Exception {
        // 0.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        // 1.读取数据
        //  sensor_1,123123213,10
        DataStreamSource<String> socketDS = env
//                .socketTextStream("localhost", 9999);
                .readTextFile("F:\\atguigu\\01_course\\code\\flink210323\\input\\sensor-data.log");

        socketDS
                .map(new MyRichMapFunction())
                .print();

        env.execute();
    }


    public static class MyRichMapFunction extends RichMapFunction<String, WaterSensor> {

        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("我 裂开 了....");
            ;
        }

        @Override
        public void close() throws Exception {
            System.out.println("我 自闭 了....");
        }

        @Override
        public WaterSensor map(String value) throws Exception {
            String[] datas = value.split(",");
            String taskName = getRuntimeContext().getTaskName();
            return new WaterSensor(
                    taskName,
                    Long.parseLong(datas[1]),
                    Integer.parseInt(datas[2])
            );
        }
    }
}




/*
    RichMapFunction： MapFunction的 富函数
        1、生命周期管理方法： open()、 close()
            open: 每个实例，一开始调用一次
            close: 每个实例，结束前调用一次
                注意，对于读文件，每个实例，close会调用 2次
        2、提供了 运行时上下文： 获取 环境的信息 、 状态、 累加器、计数器、分布式缓存等
 */